package job.qi.dftf;

import java.io.IOException;
import java.util.Iterator;

import job.qi.dftf.Qi_DFTF_Record.FieldCounter;
import mapred.base.TextReducer;
import mapred.util.TermCounter;

import org.apache.hadoop.io.Text;

public class Qi_DFTF_Count_Reducer extends TextReducer {

    @Override
    public void reduce(Text key, Iterator<Text> values) throws IOException {
        TermCounter anchor = new TermCounter();
        TermCounter title = new TermCounter();
        TermCounter body = new TermCounter();
        TermCounter query = new TermCounter();
        TermCounter url = new TermCounter();
        TermCounter all = new TermCounter();
        boolean ok = false;

        while (values.hasNext()) {
            String value = values.next().toString();
            int pos = value.indexOf('\t');
            String tag = value.substring(0, pos);
            TermCounter terms = TermCounter.parse(value.substring(pos + 1));

            if ("URL".equals(tag)) {
                url.add(terms);
                all.add(terms);
                ok = true;
            } else if ("ANCHOR".equals(tag)) {
                anchor.add(terms);
                all.add(terms);
            } else if ("TITLE".equals(tag)) {
                title.add(terms);
                all.add(terms);
            } else if ("BODY".equals(tag)) {
                body.add(terms);
                all.add(terms);
            } else if ("QUERY".equals(tag)) {
                query.add(terms);
                all.add(terms);
            }
        }

        if (ok) {
            for (String s : url) {
                this.output(new Text(s), new Text(new Qi_DFTF_Record().set(
                        FieldCounter.url_df, 1).set(FieldCounter.url_tf,
                        url.get(s)).toString()));
            }
            for (String s : title) {
                this.output(new Text(s), new Text(new Qi_DFTF_Record().set(
                        FieldCounter.title_df, 1).set(FieldCounter.title_tf,
                        title.get(s)).toString()));
            }
            for (String s : body) {
                this.output(new Text(s), new Text(new Qi_DFTF_Record().set(
                        FieldCounter.body_df, 1).set(FieldCounter.body_tf,
                        body.get(s)).toString()));
            }
            for (String s : query) {
                this.output(new Text(s), new Text(new Qi_DFTF_Record().set(
                        FieldCounter.query_df, 1).set(FieldCounter.query_tf,
                        query.get(s)).toString()));
            }
            for (String s : anchor) {
                this.output(new Text(s), new Text(new Qi_DFTF_Record().set(
                        FieldCounter.anchor_df, 1).set(FieldCounter.anchor_tf,
                        anchor.get(s)).toString()));
            }
            for (String s : all) {
                this.output(new Text(s), new Text(new Qi_DFTF_Record().set(
                        FieldCounter.all_df, 1).set(FieldCounter.all_tf,
                        all.get(s)).toString()));
            }
        }
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
